Vert.x 源码剖析 - Core

// 单机实现，创建返回VertxImpl即可
static VertxImpl vertx(VertxOptions options, Transport transport) {
    VertxImpl vertx = new VertxImpl(options, transport);
    vertx.init();
    return vertx;
}
// 集群实现，创建并加入集群
static void clusteredVertx(VertxOptions options, Transport transport, Handler<AsyncResult<Vertx>> resultHandler) {
    VertxImpl vertx = new VertxImpl(options, transport);
    vertx.joinCluster(options, resultHandler);
}

private VertxImpl(VertxOptions options, Transport transport) {
    // 创建closeHooks，CloseHooks维护了一个Closeable的Set，可向其中添加、移除任务，还有执行所有钩子的run方法啦。
    closeHooks = new CloseHooks(log);
    // 创建线程阻塞检查器，它启动一个名为vertx-blocked-thread-checker的定时器，
    checker = new BlockedThreadChecker(options.getBlockedThreadCheckInterval(), options.getBlockedThreadCheckIntervalUnit(), options.getWarningExceptionTime(), options.getWarningExceptionTimeUnit());
    // 指定一个EventLoop最长可以连续执行多久
    maxEventLoopExTime = options.getMaxEventLoopExecuteTime();
    maxEventLoopExecTimeUnit = options.getMaxEventLoopExecuteTimeUnit();
    // 创建EventLoop线程工厂，主要用于指定线程名称和线程阻塞检测器
    eventLoopThreadFactory = new VertxThreadFactory("vert.x-eventloop-thread-", checker, false, maxEventLoopExTime, maxEventLoopExecTimeUnit);
    // 创建EventLoopGroup，它又实际创建了NioEventLoopGroup，它是Netty的组件。一个EventLoopGroup，就是一个EventLoop组。在Netty中，一个EventLoop是线程和IO的结合，一个EventLoop始终绑定在同一个线程上。
    eventLoopGroup = transport.eventLoopGroup(Transport.IO_EVENT_LOOP_GROUP, options.getEventLoopPoolSize(), eventLoopThreadFactory, NETTY_IO_RATIO);
    // 创建一个acceptor EventLoopGroup，创建方式和上面类似。
    ThreadFactory acceptorEventLoopThreadFactory = new VertxThreadFactory("vert.x-acceptor-thread-", checker, false, options.getMaxEventLoopExecuteTime(), options.getMaxEventLoopExecuteTimeUnit());
    acceptorEventLoopGroup = transport.eventLoopGroup(Transport.ACCEPTOR_EVENT_LOOP_GROUP, 1, acceptorEventLoopThreadFactory, 100);
    // 创建worker线程池
    ExecutorService workerExec = new ThreadPoolExecutor(workerPoolSize, workerPoolSize,
                                                        0L, TimeUnit.MILLISECONDS, new LinkedTransferQueue<>(),
                                                        new VertxThreadFactory("vert.x-worker-thread-", checker, true, options.getMaxWorkerExecuteTime(), options.getMaxWorkerExecuteTimeUnit()));
    PoolMetrics workerPoolMetrics = metrics != null ? metrics.createPoolMetrics("worker", "vert.x-worker-thread", 	options.getWorkerPoolSize()) : null;
    workerPool = new WorkerPool(workerExec, workerPoolMetrics);
    // 创建inertnal阻塞线程池
    ExecutorService internalBlockingExec = Executors.newFixedThreadPool(options.getInternalBlockingPoolSize(),
                                                                        new VertxThreadFactory("vert.x-internal-blocking-", checker, true, options.getMaxWorkerExecuteTime(), options.getMaxWorkerExecuteTimeUnit()));
    internalBlockingPool = new WorkerPool(internalBlockingExec, internalBlockingPoolMetrics);
    // 创建文件解析器，在FileSystem中有使用，进行文件操作时使用的是java nio
    this.fileResolver = new FileResolver(options.getFileSystemOptions());
    // 创建地址解析器，在DNS解析时会用到
    this.addressResolver = new AddressResolver(this, options.getAddressResolverOptions());
    // 创建发布管理器，用于发布Verticle
    this.deploymentManager = new DeploymentManager(this);
    if (options.getEventBusOptions().isClustered()) {
        // 创建集群管理器和集群的EventBus
        this.clusterManager = getClusterManager(options);
        this.eventBus = new ClusteredEventBus(this, options, clusterManager);
    } else {
        // 创建本地EventBus
        this.clusterManager = null;
        this.eventBus = new EventBusImpl(this);
    }
    // 创建sharedData，允许你在整个应用中共享你的数据，包括集群范围内
    this.sharedData = new SharedDataImpl(this, clusterManager);
}

public EventBusImpl(VertxInternal vertx) {
    VertxMetrics metrics = vertx.metricsSPI();
    this.vertx = vertx;
    this.metrics = metrics != null ? metrics.createEventBusMetrics() : null;
    this.sendNoContext = vertx.getOrCreateContext();
}

// vertx实例时对eventBus赋值的快照
this.eventBus = new EventBusImpl(this);

@Override
public <T> MessageConsumer<T> consumer(String address, Handler<Message<T>> handler) {
    Objects.requireNonNull(handler, "handler");
    MessageConsumer<T> consumer = consumer(address);
    consumer.handler(handler);
    return consumer;
}
// 往里进一步
@Override
public <T> MessageConsumer<T> consumer(String address) {
    checkStarted();
    Objects.requireNonNull(address, "address");
    return new HandlerRegistration<>(vertx, metrics, this, address, null, false, null, -1);
}
// 重点在HandlerRegistration，收集地址后，开启超时回复定时器。
public HandlerRegistration(Vertx vertx, EventBusMetrics metrics, EventBusImpl eventBus, String address,
                               String repliedAddress, boolean localOnly,
                               Handler<AsyncResult<Message<T>>> asyncResultHandler, long timeout) {
    this.vertx = vertx;
    this.metrics = metrics;
    this.eventBus = eventBus;
    this.address = address;
    this.repliedAddress = repliedAddress;
    this.localOnly = localOnly;
    this.asyncResultHandler = asyncResultHandler;
    if (timeout != -1) {
        timeoutID = vertx.setTimer(timeout, tid -> {
            if (metrics != null) {
                metrics.replyFailure(address, ReplyFailure.TIMEOUT);
            }
            sendAsyncResultFailure(new ReplyException(ReplyFailure.TIMEOUT, "Timed out after waiting " + timeout + "(ms) for a reply. address: " + address + ", repliedAddress: " + repliedAddress));
        });
    }
}
// 最上面的consumer.handler(handler);调用了HandlerRegistration的handler方法，如下。可以看到最终是在eventBus上调用了注册方法。
@Override
public synchronized MessageConsumer<T> handler(Handler<Message<T>> h) {
    if (h != null) {
        synchronized (this) {
            handler = h;
            if (registered == null) {
                registered = eventBus.addRegistration(address, this, repliedAddress != null, localOnly);
            }
        }
        return this;
    }
    this.unregister();
    return this;
}
// 最终来到了EventBus的addRegistration方法。在addLocalRegistration中，创建了HandlerHolder，并将其加入EventBus的成员变量handlerMap，然后返回创建的HandlerHolder
protected <T> HandlerHolder<T> addRegistration(String address, HandlerRegistration<T> registration,
                                               boolean replyHandler, boolean localOnly) {
    Objects.requireNonNull(registration.getHandler(), "handler");
    LocalRegistrationResult<T> result = addLocalRegistration(address, registration, replyHandler, localOnly);
    addRegistration(result.newAddress, address, replyHandler, localOnly, registration::setResult);
    return result.holder;
}

//通过跟踪，最终会来到sendOrPubInternal，首先创建一个用于回复的HandlerRegistration，然后创建OutboundDeliveryContext，调用其next方法
public <T> void sendOrPubInternal(MessageImpl message, DeliveryOptions options,
                                  Handler<AsyncResult<Message<T>>> replyHandler) {
    checkStarted();
    HandlerRegistration<T> replyHandlerRegistration = createReplyHandlerRegistration(message, options, replyHandler);
    OutboundDeliveryContext<T> sendContext = new OutboundDeliveryContext<>(message, options, replyHandlerRegistration);
    sendContext.next();
}
// createReplyHandlerRegistration方法创建了__vertx.reply.xxx地址的响应HandlerRegistration
private <T> HandlerRegistration<T> createReplyHandlerRegistration(MessageImpl message,
                                                                  DeliveryOptions options,
                                                                  Handler<AsyncResult<Message<T>>> replyHandler) {
    if (replyHandler != null) {
        long timeout = options.getSendTimeout();
        String replyAddress = generateReplyAddress();
        message.setReplyAddress(replyAddress);
        Handler<Message<T>> simpleReplyHandler = convertHandler(replyHandler);
        HandlerRegistration<T> registration =
            new HandlerRegistration<>(vertx, metrics, this, replyAddress, message.address, true, replyHandler, timeout);
        registration.handler(simpleReplyHandler);
        return registration;
    } else {
        return null;
    }
}
protected String generateReplyAddress() {
    return "__vertx.reply." + Long.toString(replySequence.incrementAndGet());
}
// OutboundDeliveryContext类接收了消息和响应HandlerRegistration，调用next，如下。其中的iter多半是拦截器，暂时不用管。核心在sendOrPub(this)和sendReply(this, replierMessage)
@Override
public void next() {
    if (iter.hasNext()) {
        Handler<DeliveryContext> handler = iter.next();
        try {
            if (handler != null) {
                handler.handle(this);
            } else {
                next();
            }
        } catch (Throwable t) {
            log.error("Failure in interceptor", t);
        }
    } else {
        if (replierMessage == null) {
            sendOrPub(this);
        } else {
            sendReply(this, replierMessage);
        }
    }
}
// 定义io.vertx.core.eventbus.impl.EventBusImpl#sendOrPub，再定位到io.vertx.core.eventbus.impl.EventBusImpl#deliverMessageLocally,最终来到io.vertx.core.eventbus.impl.EventBusImpl#deliverMessageLocally
// 这里的关键由两个地方：一是点对点的实现——再handlerMap中找到指定地址的handlers，只取第一个进行处理；还有发布订阅的实现——对在一个地址注册的handlers全部处理；第二个关键点是消息发送的方法deliverToHandler(msg, holder)
protected ReplyException deliverMessageLocally(MessageImpl msg) {
    msg.setBus(this);
    ConcurrentCyclicSequence<HandlerHolder> handlers = handlerMap.get(msg.address());
    if (handlers != null) {
        if (msg.isSend()) {
            //Choose one
            HandlerHolder holder = handlers.next();
            if (metrics != null) {
                metrics.messageReceived(msg.address(), !msg.isSend(), isMessageLocal(msg), holder != null ? 1 : 0);
            }
            if (holder != null) {
                deliverToHandler(msg, holder);
                Handler<AsyncResult<Void>> handler = msg.writeHandler;
                if (handler != null) {
                    handler.handle(Future.succeededFuture());
                }
            }
        } else {
            // Publish
            if (metrics != null) {
                metrics.messageReceived(msg.address(), !msg.isSend(), isMessageLocal(msg), handlers.size());
            }
            for (HandlerHolder holder: handlers) {
                deliverToHandler(msg, holder);
            }
            Handler<AsyncResult<Void>> handler = msg.writeHandler;
            if (handler != null) {
                handler.handle(Future.succeededFuture());
            }
        }
        return null;
    } else {
        ... ...
    }
}
// 最终的处理函数如下：创建InboundDeliveryContext，在HandlerHolder的context环境下运行其next方法：
private <T> void deliverToHandler(MessageImpl msg, HandlerHolder<T> holder) {
    // Each handler gets a fresh copy
    MessageImpl copied = msg.copyBeforeReceive();
    DeliveryContext<T> receiveContext = new InboundDeliveryContext<>(copied, holder);

    if (metrics != null) {
        metrics.scheduleMessage(holder.getHandler().getMetric(), msg.isLocal());
    }

    holder.getContext().runOnContext((v) -> {
        try {
            receiveContext.next();
        } finally {
            if (holder.isReplyHandler()) {
                holder.getHandler().unregister();
            }
        }
    });
}
// next方法啥也没干，直接将message传入目标handler
@Override
public void next() {
    if (iter.hasNext()) {
        // ... 拦截器迭代，忽略
    } else {
        holder.getHandler().handle(message);
    }
}

// 看到只调用了一个executeAsync()
@Override
public void runOnContext(Handler<Void> task) {
    try {
        executeAsync(task);
    } catch (RejectedExecutionException ignore) {
        // Pool is already shut down
    }
}
// 这里就能看到Vertx的底了，它直接将任务提交给了netty的eventLoop
void executeAsync(Handler<Void> task) {
    nettyEventLoop().execute(() -> executeTask(null, task));
}

public synchronized Future<Void> doUndeploy(ContextInternal undeployingContext) {
    if (status == ST_UNDEPLOYED) {
        return Future.failedFuture(new IllegalStateException("Already undeployed"));
    }
    // 子发布不为空，则先取消子发布，成功后再取消当前发布。
    if (!children.isEmpty()) {
        status = ST_UNDEPLOYING;
        return doUndeployChildren(undeployingContext).compose(v -> doUndeploy(undeployingContext));
    } else {
        // 子发布为空、或取消子发布完成，现在来取消当前发布
        status = ST_UNDEPLOYED;
        List<Future> undeployFutures = new ArrayList<>();
        if (parent != null) {
            parent.removeChild(this);
        }
        // 为当前发布的每个Verticle实例执行此操作
        for (VerticleHolder verticleHolder: verticles) {
            ContextImpl context = verticleHolder.context;
            Promise p = Promise.promise();
            undeployFutures.add(p.future());
            // 该context是Verticle发布时就存好的，调用它保证了Verticle的stop和start方法在同一个线程运行。
            context.runOnContext(v -> {
                Promise<Void> stopPromise = Promise.promise();
                Future<Void> stopFuture = stopPromise.future();
                stopFuture.setHandler(ar -> {
                    // 从deployments映射中移除
                    deployments.remove(deploymentID);
                    VertxMetrics metrics = vertx.metricsSPI();
                    if (metrics != null) {
                        metrics.verticleUndeployed(verticleHolder.verticle);
                    }
                    context.runCloseHooks(ar2 -> {
                        if (ar2.failed()) {
                            // Log error but we report success anyway
                            log.error("Failed to run close hook", ar2.cause());
                        }
                        if (ar.succeeded()) {
                            p.complete();
                        } else if (ar.failed()) {
                            p.fail(ar.cause());
                        }
                    });
                });
                try {
                    // 执行Verticle的stop方法
                    verticleHolder.verticle.stop(stopPromise);
                } catch (Throwable t) {
                    if (!stopPromise.tryFail(t)) {
                        undeployingContext.reportException(t);
                    }
                }
            });
        }
        Promise<Void> resolvingPromise = undeployingContext.promise();
        CompositeFuture.all(undeployFutures).<Void>mapEmpty().setHandler(resolvingPromise);
        return resolvingPromise.future();
    }
}

private synchronized Future<Void> doUndeployChildren(ContextInternal undeployingContext) {
    if (!children.isEmpty()) {
        List<Future> childFuts = new ArrayList<>();
        // 对每个子发布执行doUndeploy方法
        for (Deployment childDeployment: new HashSet<>(children)) {
            Promise<Void> p = Promise.promise();
            childFuts.add(p.future());
            childDeployment.doUndeploy(undeployingContext, ar -> {
                children.remove(childDeployment);
                p.handle(ar);
            });
        }
        return CompositeFuture.all(childFuts).mapEmpty();
    } else {
        return Future.succeededFuture();
    }
}

private Future<Deployment> doDeploy(String identifier,
                                    DeploymentOptions options,
                                    ContextInternal parentContext,
                                    ContextInternal callingContext,
                                    ClassLoader tccl, Verticle... verticles) {
    Promise<Deployment> promise = callingContext.promise();
    String poolName = options.getWorkerPoolName();

    Deployment parent = parentContext.getDeployment();
    // 生成发布ID
    String deploymentID = generateDeploymentID();
    // 创建Deployment对象，上面有说过它是干啥的
    DeploymentImpl deployment = new DeploymentImpl(parent, deploymentID, identifier, options);

    // 发布计数
    AtomicInteger deployCount = new AtomicInteger();
    // 失败标示
    AtomicBoolean failureReported = new AtomicBoolean();
    // 如果一个Verticle发布多个实例，则会有多个verticle对象
    for (Verticle verticle: verticles) {
        // Verticle可以被要求发布到Worker线程池还是EventLoop线程池，在这里做区分
        WorkerExecutorInternal workerExec = poolName != null ? vertx.createSharedWorkerExecutor(poolName, options.getWorkerPoolSize(), options.getMaxWorkerExecuteTime(), options.getMaxWorkerExecuteTimeUnit()) : null;
        WorkerPool pool = workerExec != null ? workerExec.getPool() : null;
        // 为每个Verticle都创建一个新的Context
        ContextImpl context = (ContextImpl) (options.isWorker() ? vertx.createWorkerContext(deployment, pool, tccl) :
                                             vertx.createEventLoopContext(deployment, pool, tccl));
        if (workerExec != null) {
            context.addCloseHook(workerExec);
        }
        // 向Deployment加入Verticle对象
        deployment.addVerticle(new VerticleHolder(verticle, context));
        // 在新创建的Context上执行Verticle生命周期
        context.runOnContext(v -> {
            try {
                // 执行init方法
                verticle.init(vertx, context);
                Promise<Void> startPromise = context.promise();
                Future<Void> startFuture = startPromise.future();
                // 执行start方法
                verticle.start(startPromise);
                startFuture.setHandler(ar -> {
                    if (ar.succeeded()) {
                        if (parent != null) {
                            // 发布成功，加入父节点
                            if (parent.addChild(deployment)) {
                                deployment.child = true;
                            } else {
                                // Orphan
                                deployment.undeploy(event -> promise.fail("Verticle deployment failed.Could not be added as child of parent verticle"));
                                return;
                            }
                        }
                        // 加入发布完成的map
                        deployments.put(deploymentID, deployment);
                        // 发布的数量和待发布的数量匹配，说明发布完成，成功结束
                        if (deployCount.incrementAndGet() == verticles.length) {
                            promise.complete(deployment);
                        }
                    } else if (failureReported.compareAndSet(false, true)) {
                        // 发布失败的回滚
                        deployment.rollback(callingContext, promise, context, ar.cause());
                    }
                });
            } catch (Throwable t) {
                if (failureReported.compareAndSet(false, true))
                    deployment.rollback(callingContext, promise, context, t);
            }
        });
    }

    return promise.future();
}

public Future<Void> undeployVerticle(String deploymentID) {
    // 从deployments中获取Deployment对象
    Deployment deployment = deployments.get(deploymentID);
    // 获取当前上下文
    Context currentContext = vertx.getOrCreateContext();
    if (deployment == null) {
        return ((ContextInternal) currentContext).failedFuture(new IllegalStateException("Unknown deployment"));
    } else {
        // 调用deployment的undeploy()
        return deployment.undeploy();
    }
}